Spark & Databricks Reference

DataFrames, transformations, joins, file I/O, partitioning, SQL functions, architecture and tuning — the working set for the Databricks Certified Apache Spark Developer path.

1. Creating DataFrames

DataFrames can be built from Python collections, pandas DataFrames, or read from files. All three funnel into the same distributed structure once created.

from pyspark.sql import Row

# From a list of tuples with an explicit schema
df = spark.createDataFrame(
    [("Alice", 34), ("Bob", 45)],
    schema="name STRING, age INT"
)

# From a list of Row objects
rows = [Row(name="Alice", age=34), Row(name="Bob", age=45)]
df = spark.createDataFrame(rows)

# From a pandas DataFrame
import pandas as pd
pdf = pd.DataFrame({"name": ["Alice", "Bob"], "age": [34, 45]})
df = spark.createDataFrame(pdf)

df.printSchema()
df.show(truncate=False)

2. Selecting & Manipulating Columns

from pyspark.sql.functions import col, upper, concat, lit

df.select("name", "age").show()
df.select(col("name"), (col("age") + 1).alias("age_next_year")).show()

# Renaming
df.withColumnRenamed("age", "years_old")

# Adding / transforming columns
df.withColumn("name_upper", upper(col("name"))) \
  .withColumn("greeting", concat(lit("Hi "), col("name")))

3. Filtering, Dropping, Sorting

# Filtering rows
df.filter(col("age") > 30)
df.where((col("age") > 30) & (col("name") != "Bob"))

# Dropping columns / rows
df.drop("age")
df.dropna(subset=["name"])
df.dropDuplicates(["name"])

# Sorting
df.orderBy(col("age").desc())
df.sort("name", ascending=True)

4. Aggregations & Joins

from pyspark.sql.functions import avg, count, max as smax

df.groupBy("department").agg(
    count("*").alias("headcount"),
    avg("age").alias("avg_age"),
    smax("age").alias("oldest")
)

# Joins
df1.join(df2, on="id", how="inner")
df1.join(df2, df1.dept_id == df2.id, how="left")
df1.join(df2, "id", how="left_semi")   # keep df1 rows with a match
df1.join(df2, "id", how="left_anti")   # keep df1 rows with no match

5. Reading & Writing Data

Every Spark job starts with spark.read and usually ends with df.write. Most of the time the short form below is enough, but the options you do not set decide how fast the read is, what happens to bad rows, and whether a failed job loses data.

# Reading
df = spark.read.csv("/data/orders.csv", header=True, inferSchema=True)
df = spark.read.json("/data/events.json")
df = spark.read.parquet("/data/sales/")

# Writing
df.write.mode("overwrite").parquet("/data/out/")
df.write.mode("append").partitionBy("year", "month").parquet("/data/out/")
df.write.format("delta").save("/data/delta/orders")

This section follows a file through a job: how the reader works, setting the schema, handling bad rows, choosing files, then reading from databases. The second half covers writing: what Spark produces, save modes, the overwrite danger, compression, and what changes with and without Databricks.

How the reader and writer work

Both follow the same pattern: name a format, add options, then give a path or table. spark.read.csv(path) is only a shortcut for the longer form.

df = (spark.read
        .format("csv")  # csv, json, parquet, orc, text, jdbc, delta
        .schema(order_schema)          # column names and types
        .option("header", True)        # one setting
        .options(sep=",", quote='"')   # several settings
        .load("/data/orders/"))

(df.write
   .format("parquet")
   .mode("append")                     # what to do if data exists
   .option("compression", "zstd")
   .partitionBy("order_date")          # folders on disk
   .save("/data/out/orders"))

Reading is lazy. spark.read returns a plan and reads no rows until an action such as count() or a write runs. The exception is schema inference, which must look at the data straight away, as the next part shows.

Step 1: define the schema

The schema is the most important read option. There are three ways to set it:

from pyspark.sql.types import (StructType, StructField, IntegerType,
                               StringType, DoubleType, DateType)

# Best to read: a DDL string
schema = "id INT, name STRING, amount DOUBLE, order_date DATE"

# The same thing as a StructType
schema = StructType([
    StructField("id", IntegerType(), nullable=False),
    StructField("name", StringType()),
    StructField("amount", DoubleType()),
    StructField("order_date", DateType()),
])

df = spark.read.schema(schema).option("header", True).csv(path)

# Reuse a known schema from a sample
schema = spark.read.parquet(sample_path).schema

We read a 144 MB CSV of 3 million rows both ways:

The cost grows with the data, but the bigger risk is a wrong guess. With inferSchema, the zip codes 01234 and 00501 became the integers 1234 and 501, and the leading zeros were gone for good. A column of numbers can also turn into strings the day one bad value arrives. Use inferSchema for exploring, and an explicit schema in pipelines.

Step 2: decide what happens to bad rows

Real files contain broken rows. For CSV and JSON, the mode option decides what Spark does with a row that does not match the schema.

In a pipeline, the best pattern is usually to split good rows from bad rows and keep the bad ones for review. Add a string column to the schema, and Spark stores the raw text of each bad row in it. Cache first: without it, selecting only this column returned NULL for every row in our test, even though the bad row existed.

schema = "id INT, name STRING, amount DOUBLE, _corrupt_record STRING"

raw = (spark.read.schema(schema)
         .option("header", True)
         .option("columnNameOfCorruptRecord", "_corrupt_record")
         .csv(path)
         .cache())

good = raw.filter("_corrupt_record IS NULL").drop("_corrupt_record")
bad  = raw.filter("_corrupt_record IS NOT NULL")

bad.write.mode("append").json("/quarantine/orders/")

The count() trap

Spark reads only the columns a query needs, which also applies to deciding whether a row is bad. We read 3 rows, one with abc in the amount column, and got different answers from the same DataFrame. The rule: a bad value is only noticed if the query really reads that column.

So count() cannot prove a file is clean, and even select("amount").count() is optimized away. Setting spark.sql.csv.parser.columnPruning.enabled to false did not change this result. Compute something on the column, write the data out, or use the split pattern above.

Options for each format

nullValue is worth setting. In our test, with nullValue set to NA, the value NA became a clean NULL instead of being flagged as a bad row.

JSON expects one object per line by default. A normal pretty-printed file with one big array is not in that form. Reading ours without multiLine gave 4 rows and a _corrupt_record column, with no error at all; with it, we got the correct 2 rows.

spark.read.json(path)  # one object per line
spark.read.option("multiLine", True).json(path)  # one array or object

# Nested fields use dot notation
df.select("id", "address.city")

Parquet and ORC store the schema in each file, so they need no schema option and few other options. They also let Spark skip work. For a filter on one column, the physical plan showed all three of these:

Parquet does have one trap. If files have different columns, Spark reads the schema from only one of them by default, so columns can silently disappear. In our test, the extra column was missing until we set mergeSchema. It has a cost, since Spark must read every file footer.

spark.read.option("mergeSchema", True).parquet(path)

spark.read.text(path)                     # one column: value
spark.read.format("binaryFile").load(path)  # path, length, content...
spark.read.orc(path)

Choosing which files to read

spark.read.parquet("/sales/")  # a whole folder
spark.read.parquet("/sales/y=2026/m=*")  # a glob pattern
spark.read.option("pathGlobFilter", "*.parquet")  # match file names
spark.read.option("recursiveFileLookup", True)  # look in subfolders
spark.read.option("modifiedAfter", "2026-10-01T00:00:00")  # recent files

# Which file did each row come from?
df.select("*", F.col("_metadata.file_path"))

Partition folders behave in ways that surprise people. We tested these on a table partitioned by y and m:

# Read a few folders and keep the partition columns
(spark.read.option("basePath", "/sales")
      .parquet("/sales/y=0", "/sales/y=1"))

_metadata is hidden from select("*"), so name it explicitly. modifiedAfter matching no files gave an "unable to infer schema" error in our test, so pass a schema when no files is a normal outcome.

Reading from a database with JDBC

By default, a JDBC read uses one connection and one partition, however large the table is. To read in parallel, tell Spark how to split it.

import os

orders = (spark.read.format("jdbc")
    .option("url", "jdbc:postgresql://db-host:5432/shop")
    .option("user", "etl_user")
    .option("password", os.environ["DB_PASSWORD"])
    .option("dbtable", "orders")
    .option("partitionColumn", "id")    # number, date or timestamp
    .option("lowerBound", 1)            # all four go together
    .option("upperBound", 1000000)
    .option("numPartitions", 8)         # also the connection limit
    .option("fetchsize", 10000)         # rows per round trip
    .load())

lowerBound and upperBound are widely misunderstood. They do not filter rows; they only set how the range is cut. Rows outside the range still arrive, in the first or last partition. We read 1,000 rows with 4 partitions:

Wrong bounds silently create skew, and one connection does nearly all the work. Get the real range first:

bounds = (spark.read.format("jdbc").option("url", url)
    .option("query", "SELECT MIN(id) AS lo, MAX(id) AS hi FROM orders")
    .load().first())

# query cannot be combined with partitionColumn, so use a subquery
sub = "(SELECT id, amount FROM orders WHERE status = 'PAID') t"
df = (spark.read.format("jdbc").option("url", url)
    .option("dbtable", sub)
    .option("partitionColumn", "id")
    .option("lowerBound", bounds.lo).option("upperBound", bounds.hi)
    .option("numPartitions", 8)
    .load())

Spark pushes simple filters to the database: the plan for a filter on id showed PushedFilters on the JDBC scan. Each partition opens a real connection, so do not overload a production database. Read from a replica if you can, and use the database for the query, not a Spark filter on a whole table.

What df.write actually produces

Spark does not write a file; it writes a folder. Each partition writes its own file, and coalesce(1) gives you one data file, still inside a folder.

/data/out/orders/
  _SUCCESS                                   empty: the job finished
  part-00000-a1b2...-c000.snappy.parquet     one per partition
  part-00001-a1b2...-c000.snappy.parquet
  part-00002-a1b2...-c000.snappy.parquet

So the number of partitions when you write sets the number of files. See Partition Optimization for how to choose it.

Save modes

ignore reports success while writing nothing, so a job can look healthy when it did no work. append is not idempotent: rerunning a job duplicates its rows.

The overwrite danger on plain files

This is the most important thing in this section. On plain Parquet or CSV, overwrite deletes the existing files first and writes the new ones afterwards, so there is a window with no data. Two everyday situations turn that window into data loss. We tested both on 1,000 rows:

The first case happens because Spark is lazy: the delete runs before the read has started, so the files it needs have already gone, and you see a FileNotFoundException. Delta avoids both, because an overwrite only commits a new version to its transaction log, and old files stay until the commit succeeds.

# DANGEROUS on plain files: reads and deletes the same path
df = spark.read.parquet(path).filter("amount > 0")
df.write.mode("overwrite").parquet(path)

# Safe: write elsewhere, check it, then swap
tmp = path + "_tmp"
df.write.mode("overwrite").parquet(tmp)
assert spark.read.parquet(tmp).count() > 0
spark.read.parquet(tmp).write.mode("overwrite").parquet(path)

# Better: use Delta, where overwrite is atomic
df = spark.read.format("delta").load(delta_path).filter("amount > 0")
df.write.format("delta").mode("overwrite").save(delta_path)

A cache() made the first case work in our test, but it is not a safe fix. Cached partitions can be evicted and re-read from files that no longer exist. The swap above still has a short gap, so it reduces the risk but does not remove it. Only a table format such as Delta removes it.

Overwriting only part of a table

There is a second overwrite trap. With partitionBy, the default overwrite replaces the whole table, not only the partitions in your DataFrame. We overwrote 4 partitions with data for one, and only that one folder remained.

# Plain files: replace only the partitions present in df
(df.write.mode("overwrite")
   .option("partitionOverwriteMode", "dynamic")   # 4 folders stayed
   .partitionBy("month")
   .parquet(path))

# Delta: replace rows that match a condition
(df.write.format("delta").mode("overwrite")
   .option("replaceWhere", "month = 2")
   .save(delta_path))

replaceWhere is also a safety check. When the DataFrame contained rows outside the condition, Delta rejected the write with DELTA_REPLACE_WHERE_MISMATCH.

Schema changes on write

Plain files do not protect you here. Our append succeeded, and then reading the folder showed only the old columns, which is the same silent schema problem described under Parquet above.

Compression and file format

The same 3 million rows, written once in each form:

df.write.option("compression", "zstd").parquet(path)
df.write.option("compression", "gzip").option("header", True).csv(path)

Sizes depend heavily on the data, so test yours. But one rule is reliable: gzip files cannot be split. One 38 MB plain CSV was read as 4 partitions; the same data as one gzip file was read as 1 partition, whatever its size. A large gzip file therefore means one slow task. Parquet is different, as it compresses inside the file and stays splittable.

CSV cannot hold nested data: writing an array column failed with UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE. Avro needs the separate spark-avro package on open-source Spark.

Writing to files or to tables

The Data Modelling guide tests these on Delta: insertInto silently stored a surname as the first name, while saveAsTable matched by name.

Writing to a database with JDBC

(df.write.format("jdbc")
   .option("url", url)
   .option("dbtable", "reporting.daily_sales")
   .option("user", "etl_user")
   .option("password", os.environ["DB_PASSWORD"])
   .option("batchsize", 5000)        # rows per insert
   .option("numPartitions", 4)       # at most 4 connections
   .mode("append")
   .save())

# Empty the table but keep its indexes and grants
(df.write.format("jdbc").option("url", url)
   .option("dbtable", "daily_sales")
   .option("truncate", True)
   .mode("overwrite").save())

Plain overwrite drops and recreates the database table, which loses indexes and grants. truncate avoids that, though PostgreSQL does not support it and the option is ignored there. Spark has no JDBC upsert, so for that write to a staging table and run a MERGE in the database.

Without Databricks: open-source PySpark

You are responsible for paths, safety and file sizes. Delta needs to be added to the session, and the default format is Parquet.

# Local files, S3 or ADLS paths
df = spark.read.schema(schema).option("header", True).csv("s3a://raw/orders/")

# Without a format, Spark assumes Parquet
df.write.mode("append").save("s3a://curated/orders/")

# Delta needs the package, and the format named
df.write.format("delta").mode("append").save("s3a://lake/orders/")

On our local Spark, saveAsTable created a Parquet table by default, not Delta. Do not assume Databricks behaviour when moving code.

With Databricks

The API is the same, but defaults and tools differ. Delta is the default format, tables live in Unity Catalog, and raw files usually sit in a volume.

# Delta is the default, so no format is needed
df = spark.read.table("main.sales.orders")
df.write.mode("append").saveAsTable("main.sales.orders_clean")

# Files in a Unity Catalog volume
df = (spark.read.schema(schema).option("header", True)
        .csv("/Volumes/main/sales/raw/orders/"))

# Secrets instead of passwords in code
password = dbutils.secrets.get("etl", "db_password")

For files, Databricks adds read_files, which detects the format, infers one schema across all files, and captures anything that does not fit in a rescued data column rather than failing.

-- Batch
SELECT *, _metadata.file_path
FROM read_files('/Volumes/main/sales/raw/orders/',
                format => 'csv', header => true);

-- Incremental: a streaming table uses Auto Loader
CREATE OR REFRESH STREAMING TABLE bronze_orders AS
SELECT * FROM STREAM read_files('/Volumes/main/sales/raw/orders/');

With and without Databricks, side by side

Common mistakes

Common interview questions

6. Partitioning

Partitioning controls how data is split across tasks and, for storage, across directories.

df.rdd.getNumPartitions()
df.repartition(8)                 # shuffle to exactly 8 partitions
df.repartition(8, "region")       # shuffle, partitioned by key
df.coalesce(2)                    # narrow transformation, merges partitions, no shuffle

df.write.partitionBy("year", "month").parquet("/data/out/")

coalesce avoids a full shuffle and is cheaper for reducing partitions; repartition is needed to increase partitions or rebalance skewed data.

To work out how many partitions to use, see Partition Optimization.

7. Spark SQL Functions

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number

w = Window.partitionBy("department").orderBy(col("salary").desc())
df.withColumn("rank_in_dept", row_number().over(w))

8. Spark Architecture

• Driver — runs the user's main program, builds the DAG of transformations, and schedules tasks.
• Cluster manager — (YARN, Kubernetes, or Spark's standalone manager) allocates resources for executors. In standalone mode a Spark Master JVM plays this role, with Spark workers registering into its quorum.
• Executors — JVM processes on worker nodes that run tasks and cache data; each executor exposes a fixed number of slots that run tasks concurrently.
• Job → Stages → Tasks — an action (like .collect()) triggers a job; the DAG scheduler splits it into stages at shuffle boundaries; each stage runs one task per partition.
• Lazy evaluation — transformations (select, filter, join) build a plan; nothing runs until an action (show, count, write, collect).
• Catalyst optimizer — rewrites the logical plan (predicate pushdown, constant folding) before generating a physical plan.
• Tungsten — off-heap binary memory format and whole-stage code generation for CPU efficiency.

9. What Happens on spark-submit (End to End)

This walkthrough follows one job from start to finish. Each step happens because of the one before it, so read it as a story rather than a list of facts.

The job

df     = spark.read.parquet("s3://lake/sales/")  # 10 GB of files
clean  = df.filter(F.col("amount") > 0).select("region", "amount")
totals = clean.groupBy("region").agg(F.sum("amount"))
totals.write.parquet("s3://lake/gold/region_totals/")

spark-submit --master yarn --deploy-mode cluster \
  --num-executors 10 --executor-cores 5 --executor-memory 16g sales_job.py

The story

1. You run spark-submit. It does not run your code itself. It asks the cluster manager (YARN) for a place to run a driver, because every Spark application is led by one driver process.

2. YARN starts the driver. The ResourceManager picks a machine and starts a container there. In cluster mode the driver lives inside that container, so your laptop can now disconnect.

3. The driver's first job is to get workers. It cannot process 10 GB alone, so it asks YARN for the executors you requested on the command line: 10 executors × 5 cores × 16 GB. YARN finds machines with free capacity and starts an executor JVM on each.

4. The executors report back. Each executor registers with the driver: "I'm executor 3, I have 5 cores." The driver now knows it has 10 × 5 = 50 cores, meaning it can run 50 tasks at the same time. Remember this number: it decides how fast step 10 goes.

5. Now the driver runs your Python code, line by line. The first three lines (read, filter/select, groupBy) are transformations, and Spark does not execute transformations. It only writes them down as a plan: "read sales → keep amount > 0 → keep 2 columns → group by region → sum". No data has been read yet; the cluster is idle.

6. Why wait? So the plan can be optimized as a whole. Because Spark sees the entire chain before running anything, its optimizer (Catalyst) can rewrite it: push amount > 0 down into the file read so fewer rows are loaded, read only the region and amount columns from Parquet, and plan the group-by as partial sums first, then a final sum. None of this would be possible if each line ran immediately.

7. The fourth line, write, is an action, and an action forces execution. You have asked for a real result (files in S3), so the driver must turn the plan into actual work. That work is a job.

8. The driver splits the job into stages, because of the groupBy. To sum per region, every row for "North" must end up in the same place, but right now those rows are scattered across all the files. Moving rows between machines by key is a shuffle, and a shuffle cannot happen in the middle of a flow. So the driver cuts the job in two: Stage 0 does everything that can happen where the data already is (read → filter → select → partial sum per region); Stage 1 does everything after the rows are regrouped (final sum → write). Stage 1 cannot start until Stage 0 has finished everywhere, because it needs every piece's partial sums.

9. Each stage is split into tasks, one per piece of data. The 10 GB input is split into roughly 128 MB pieces called partitions, about 80 of them. One task handles one partition, so Stage 0 = 80 tasks. Each task runs the whole Stage 0 pipeline for its own piece: read its 128 MB, filter, select and partial-sum, in a single pass.

10. The driver hands tasks to the cores from step 4. There are 80 tasks but only 50 cores, so the first 50 run immediately; as each finishes, its core picks up one of the remaining 30. The driver prefers to send a task to an executor that already has its data nearby.

11. Stage 0 tasks finish by preparing for the shuffle. Each task writes its partial sums to the executor's local disk, sorted into buckets by region (shuffle files), and tells the driver where they are.

12. Once all 80 are done, Stage 1 begins. Stage 1 tasks fetch those shuffle files over the network, so each task gathers every partial sum for its regions, adds them up and writes its part of the output to S3. The number of Stage 1 tasks comes from the shuffle-partition setting (200 by default); Adaptive Query Execution (AQE) sees there are only a few regions and merges them down to a handful of tasks.

13. The last Stage 1 task finishes, so the job is done. The output is in S3. If the script contains another action, steps 7–12 repeat for it: a new job, new stages and new tasks.

14. Your script ends, so everything shuts down. The driver releases the executors, YARN frees the containers, and the application is finished.

And if something fails along the way? If a task fails, the driver simply re-runs it (up to 4 times). If an executor dies and Stage 0's shuffle files are lost, the driver still has the plan from step 5, so it re-runs only the Stage 0 tasks needed to rebuild the missing files. That is why Spark keeps the plan (the "lineage") instead of copying data for safety.

The chain in one picture

spark-submit
   └─► YARN starts the DRIVER
          └─► driver requests EXECUTORS → 10 × 5 cores = 50 task slots
                 └─► your code runs: transformations only BUILD A PLAN (nothing executes)
                        └─► optimizer improves the whole plan
                               └─► ACTION (write) → plan becomes a JOB
                                      └─► groupBy needs a SHUFFLE → job splits into 2 STAGES
                                             └─► each stage splits into TASKS (1 per partition: 80)
                                                    └─► tasks run on the 50 cores (2 waves)
                                                           └─► Stage 0 writes shuffle files
                                                                  └─► Stage 1 reads them, sums, writes S3
                                                                         └─► job done → app ends

Every step exists because of the one before it: executors because the driver cannot do the work alone; lazy transformations so the optimizer sees the whole plan; the action turns the plan into a job; stages because the shuffle forces a pause; tasks because the data is split into partitions; and the run time depends on how many tasks there are compared with the cores from step 4.

Quick reference

AQE detail, measured on Spark 3.5 with this exact pattern: with Adaptive Query Execution off, the write ran as 1 job with 2 stages (4 map tasks, then 8 shuffle tasks). With AQE on (the default since Spark 3.2), Spark ran each stage as its own job so it could re-plan with real statistics: 2 jobs, the second showing the map stage as skipped and the reduce stage coalesced from 8 partitions to 1 task. So in the Spark UI one action can appear as several jobs.

10. Performance Tuning

• Broadcast joins — broadcast small tables (default threshold 10MB, spark.sql.autoBroadcastJoinThreshold) to avoid a shuffle join.
• Adaptive Query Execution (AQE) — spark.sql.adaptive.enabled=true re-optimizes plans at runtime: coalesces small shuffle partitions, splits skewed ones, and can switch join strategies.
• Shuffle partitions — spark.sql.shuffle.partitions (default 200) should scale with data volume and cluster size.
• Caching — df.cache() / .persist(StorageLevel.MEMORY_AND_DISK) for DataFrames reused across multiple actions.
• File sizing — avoid the small-files problem; prefer a handful of well-sized Parquet files per partition over thousands of tiny ones.
• Data skew — salting keys or enabling AQE's skew join handling when a few keys dominate a shuffle.
• Column pruning & predicate pushdown — select only needed columns and filter early; columnar formats (Parquet/Delta) push both down to the file scan.

Spotting data skew

Skew means a few keys hold most of the rows. After a shuffle, every row for a key lands in the same partition, so one task processes far more data than the others and the whole stage waits for it. In the Spark UI, open the slow stage and compare task metrics: a max duration or shuffle-read size far above the median (e.g. 20 minutes vs 30 seconds) is skew. Confirm by counting rows per key:

from pyspark.sql import functions as F

(txn.groupBy("customer_id").count()
    .orderBy(F.desc("count"))
    .show(10))          # a handful of keys (often NULL or a default like 'UNKNOWN') dominate

Try the cheaper fixes first: broadcast the small side if it fits in memory (no shuffle, so no skew); let AQE skew-join handling split oversized partitions (spark.sql.adaptive.skewJoin.enabled, on by default with AQE in Spark 3.2+); and handle NULL keys separately, since they never match in a join anyway. Reach for salting when those are not enough: extreme skew, join types or sides AQE cannot split, or skewed aggregations and windows.

Salting a skewed join

Salting spreads one hot key over N partitions. Add a random salt (0 to N−1) to each row on the large, skewed side, replicate every row of the other side N times (once per salt value), and join on key + salt. Each original match still happens exactly once, but the hot key's rows are now split across N tasks.

from pyspark.sql import functions as F

SALT_BUCKETS = 16

# 1. Large, skewed side: a random salt per row
txn_salted = txn.withColumn("salt", (F.rand(seed=42) * SALT_BUCKETS).cast("int"))

# 2. Other side: one copy of every row per salt value
salts = spark.range(SALT_BUCKETS).select(F.col("id").cast("int").alias("salt"))
cust_salted = customers.crossJoin(salts)

# 3. Join on the original key plus the salt, then drop the helper column
joined = (txn_salted
          .join(cust_salted, on=["customer_id", "salt"], how="inner")
          .drop("salt"))

Replicating the whole small side N times can be wasteful. A refinement salts only the hot keys and leaves every other key with salt 0:

hot_keys = [r.customer_id for r in
            txn.groupBy("customer_id").count()
               .filter(F.col("count") > 1_000_000)
               .select("customer_id").collect()]          # small list, safe to collect

is_hot = F.col("customer_id").isin(hot_keys)

txn_salted = txn.withColumn(
    "salt",
    F.when(is_hot, (F.rand(seed=42) * SALT_BUCKETS).cast("int")).otherwise(F.lit(0)))

cust_salted = (customers.filter(is_hot).crossJoin(salts)            # hot keys: N copies
               .unionByName(customers.filter(~is_hot)
                                     .withColumn("salt", F.lit(0)))) # others: one copy

joined = txn_salted.join(cust_salted, ["customer_id", "salt"]).drop("salt")

• Choosing N: roughly the hot key's size divided by a healthy partition size (100–200 MB). Larger N spreads more but multiplies the replicated side; 8–64 is typical.
• Join types: works for inner joins and for outer joins that preserve the salted (large) side, e.g. txn LEFT JOIN customers. Do not salt the side whose unmatched rows must be preserved, or each unmatched row would appear N times.
• Use a fixed seed so reruns and task retries assign the same salts.

Salting a skewed aggregation (two-stage)

Aggregate on key + salt first, so the hot key is pre-aggregated across N tasks, then aggregate the partial results on the key alone:

partial = (txn
    .withColumn("salt", (F.rand(seed=42) * SALT_BUCKETS).cast("int"))
    .groupBy("customer_id", "salt")
    .agg(F.sum("amount").alias("amount"),
         F.count("*").alias("txns")))

result = (partial
    .groupBy("customer_id")
    .agg(F.sum("amount").alias("total_amount"),
         F.sum("txns").alias("txns"),
         (F.sum("amount") / F.sum("txns")).alias("avg_amount")))   # rebuild AVG from SUM / COUNT

• Spark already performs map-side partial aggregation for simple functions like sum, count, min and max, so plain aggregations rarely need this. Two-stage salting pays off for functions whose partial state stays large, such as collect_list / collect_set, percentiles, or heavy custom aggregations.
• The second stage must be able to combine partial results: SUM of sums, SUM of counts, MAX of maxes. AVG must be rebuilt from SUM and COUNT, and distinct counts cannot be split this way.

spark-submit template

A production-style submit for a PySpark job on YARN. Everything after the script path is passed to your script as arguments.

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --name daily-txn-enrichment \
  --driver-memory 4g \
  --driver-cores 2 \
  --num-executors 10 \
  --executor-cores 5 \
  --executor-memory 16g \
  --conf spark.executor.memoryOverhead=2g \
  --conf spark.sql.shuffle.partitions=400 \
  --conf spark.sql.adaptive.enabled=true \
  --conf spark.sql.adaptive.skewJoin.enabled=true \
  --conf spark.sql.autoBroadcastJoinThreshold=50m \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.dynamicAllocation.minExecutors=2 \
  --conf spark.dynamicAllocation.maxExecutors=20 \
  --conf spark.dynamicAllocation.shuffleTracking.enabled=true \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.1 \
  --py-files dist/common_utils.zip \
  --files config/job.yaml \
  jobs/enrich_txn.py --run-date 2026-10-25 --env prod

Inside the job, read the arguments with argparse. SQL settings can also be changed at runtime, but resource settings (memory, cores, executor count) must be fixed before the application starts:

import argparse
from pyspark.sql import SparkSession

parser = argparse.ArgumentParser()
parser.add_argument("--run-date", required=True)
parser.add_argument("--env", default="dev")
args = parser.parse_args()

spark = SparkSession.builder.appName("daily-txn-enrichment").getOrCreate()
spark.conf.set("spark.sql.shuffle.partitions", "400")   # SQL configs: OK at runtime
# spark.executor.memory cannot be changed here; set it in spark-submit or cluster config

Run the same job locally with spark-submit --master local[4] jobs/enrich_txn.py --run-date 2026-10-25 --env dev. On Databricks there is no spark-submit for normal jobs: the cluster or job definition sets these values.

11. Partition Optimization

A partition is a chunk of data, and one task processes one partition on one core. Partition optimization means choosing how many partitions you have, and how large they are, at each point in a job, so that every core stays busy without any task being overloaded and without flooding storage with tiny files.

This section builds up in order: where partitions are decided, how to calculate the number, which tools change it, what the measurements show, and what differs with and without Databricks.

Partitions are decided in four places

"Number of partitions" is not one setting. A single job goes through four stages, and the first three are about splitting work in memory, while the last is about folders on disk. Confusing these is the most common source of mistakes.

  Files in storage
        │  1. READ: split files into input partitions
        ▼
  Narrow work (filter, select, withColumn): count unchanged
        │  2. SHUFFLE: groupBy, join, orderBy, distinct
        ▼
  Shuffle partitions
        │  3. WRITE: one file per partition, per folder
        ▼
  Output files
        │  4. STORAGE LAYOUT: partitionBy folders
        ▼
  year=2026/month=10/part-0001.parquet

Reading usually sorts itself out, because Spark splits files by size. So in practice the decision comes down to shuffle partitions and output files.

How to decide the number of partitions

There is no universal number; you work it out from three facts about your job: how much data moves, how large a partition should be, and how many cores you have. Follow one job from start to finish: it reads 100 GB, on 10 executors with 5 cores and 16 GB each, so 50 cores.

1. Find out how much data is shuffled. The input size is not the number that matters, because filtering and column pruning happen first. Run the job once and read Shuffle Write on the stage page of the Spark UI. Ours filters 100 GB down to 60 GB of shuffle data.

2. Divide by a target partition size. A common starting point is 100 to 200 MB per shuffle partition. At 200 MB: 60 GB ÷ 200 MB ≈ 308 partitions.

3. Check against the cores. You want more tasks than cores, so that when one task runs long the others pick up the work. Spark's tuning guide suggests 2 to 3 tasks per core. For 50 cores that means at least 100 to 150, and 308 passes. If data is small, this rule sets the number instead of step 2.

4. Check that a partition fits in memory. Each executor has about 9.4 GB of unified memory and runs 5 tasks, so each task gets at most about 1.9 GB. A 200 MB partition fits easily. If yours did not, you would need more partitions or fewer cores per executor.

5. Decide the output files separately. If the job writes 20 GB, aim for files of about 256 MB, so about 80 files, using repartition(80) before writing. This has nothing to do with the 308 used for shuffling.

6. Run it, then check the Spark UI. The calculation only gives you a starting value. The table below tells you which way to adjust.

import math

def shuffle_partitions(shuffle_gb, total_cores, target_mb=200):
    """Starting value: data / target size, but keep every core busy."""
    by_size = math.ceil(shuffle_gb * 1024 / target_mb)
    by_cores = total_cores * 3
    return max(by_size, by_cores)

cores = 10 * 5  # 10 executors, 5 cores each
print(shuffle_partitions(60, cores))           # 308
print(shuffle_partitions(2, cores))            # 150

As a quick reference, with 200 MB partitions:

Reading the Spark UI to confirm

The tools for changing partitions

df.rdd.getNumPartitions()                      # how many now?

df.repartition(80)                             # exactly 80, evenly
df.repartition(80, "customer_id")  # same key, same partition
df.coalesce(10)  # merge without a shuffle
df.repartitionByRange(50, "order_date")        # date ranges

# AQE evens out the output sizes
spark.sql("SELECT /*+ REBALANCE */ * FROM sales")

# Cap each output file
df.write.option("maxRecordsPerFile", 5_000_000).parquet(path)

# See how evenly rows are spread across partitions
(df.withColumn("pid", F.spark_partition_id())
   .groupBy("pid").count()
   .orderBy(F.desc("count")).show(5))

coalesce moves no data, so it merges whole partitions, and it also reduces the parallelism of the work before it. In our test, coalesce(1) ran the whole write as 1 task, while repartition(1) ran 4 tasks and then 1. Use repartition if the steps before it are expensive, or if partitions are badly uneven.

Disk partitions: what partitionBy really does

partitionBy is the fourth stage, and it is a different thing. It creates a folder per value so queries can skip whole folders, which is called partition pruning. Its danger is the multiplication: Spark writes up to one file per in-memory partition, per folder.

files written  ≈  in-memory partitions  x  folders each one touches

# 16 random partitions x 30 dates: up to 480 small files
df.repartition(16).write.partitionBy("day").parquet(path)

# Each day goes to one partition first: 30 files
df.repartition("day").write.partitionBy("day").parquet(path)

Choose a column that queries filter on, with few distinct values: date or region, not customer_id. A column with millions of values creates millions of folders.

Measured: what partition counts actually do

These ran on Spark 3.5 with 4 local cores. The first table is a group-by and sort of 5 million rows, using the median of 3 runs, with AQE off so that the setting is the only change.

Output and storage choices, on 10 million rows and then 3 million rows. These were single runs, so read small gaps as noise:

A filter on the partition column took 0.15 s against 0.33 s on a normal column of the same table. Lowering maxPartitionBytes from 128 MB to 8 MB raised the read from 4 to 28 partitions, which shows how Spark splits by size.

• Too many partitions costs more than too few here. One partition cost 0.4 s more than the best; 4,000 cost 23 s more. On this small dataset, overhead dominated.
• The default of 200 is not always right. It is a fixed number, chosen with no knowledge of your data. It was too high here, and it would be far too low for 1 TB.
• AQE removes most of the risk. Starting high and letting AQE merge gave nearly the best result, which is why that is the standard approach.
• Repartitioning has a real cost. Each repartition adds a shuffle, so do it only for a reason, such as controlling output files.

Small data on one machine, so a cluster moves the best value: more data needs more partitions. coalesce(1) and repartition(1) took about the same time after an expensive step here, at 11 to 15 s each, so neither clearly won. The direction is what matters, so measure your own job.

Without Databricks: open-source PySpark

On plain Spark, whether on EMR, Dataproc, YARN or Kubernetes, you make these choices yourself. The usual approach is to start high and let AQE merge partitions.

from pyspark.sql import SparkSession

spark = (SparkSession.builder
    .appName("sales_job")
    # 1. Read: size of an input split
    .config("spark.sql.files.maxPartitionBytes", "134217728")

    # 2. Shuffle: start high, and let AQE merge down
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
    .config("spark.sql.shuffle.partitions", "1000")

    # Target size after merging, and respect it on a busy cluster
    .config("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m")
    .config("spark.sql.adaptive.coalescePartitions.parallelismFirst",
            "false")

    # Skew: split oversized partitions in joins
    .config("spark.sql.adaptive.skewJoin.enabled", "true")
    .getOrCreate())

Setting names above are shortened; each starts with spark.sql.adaptive. Leaving parallelismFirst at true means AQE ignores your target size. The Spark docs suggest false on a busy cluster, to avoid many small tasks.

Writing and storage are also your job on plain Spark:

# Good: one write per day, and one folder per day
(df.repartition("order_date")
   .write.mode("overwrite")
   .option("maxRecordsPerFile", 5_000_000)
   .partitionBy("order_date")
   .parquet(path))

# Rewrite only the days present in this DataFrame, not the whole table
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

# Compact one day's small files
day = f"{path}/order_date=2026-10-30"
tmp = f"{path}_tmp/order_date=2026-10-30"
spark.read.parquet(day).repartition(4).write.mode("overwrite").parquet(tmp)
# then swap tmp into place in storage

With plain Parquet nothing compacts files for you, so schedule it. With Delta Lake or Iceberg outside Databricks, use their compaction commands, such as Delta's OPTIMIZE. For choosing partitionBy columns, the size rule below is still a sensible guide.

With Databricks

Databricks automates much of this, so the advice changes: start by removing manual settings, not adding them.

-- Shuffle: let Databricks choose from the query plan and input size
SET spark.sql.shuffle.partitions = auto;

-- Layout: use clustering for new tables, not partitioning
CREATE TABLE prod.gold.fact_sales (
  order_id STRING, order_date DATE, region STRING, amount DECIMAL(18,2)
) CLUSTER BY (order_date, region);

-- Or let Databricks choose the keys (Unity Catalog managed tables)
CREATE TABLE prod.gold.fact_returns (order_id STRING, return_date DATE)
CLUSTER BY AUTO;

-- Merge small files and apply clustering
OPTIMIZE prod.gold.fact_sales;

Should a Delta table on Databricks be partitioned?

Liquid clustering cannot be combined with partitioning or ZORDER on the same table. For stateful streaming queries, the shuffle partition count is fixed by the checkpoint, so choose it before the first run.

With and without Databricks, side by side

The calculation stays useful in both. On Databricks it tells you whether a result looks reasonable and whether a fixed value makes sense; without Databricks, it sets the value you run with.

Common mistakes

Common interview questions

12. Tuning Questions from the Official Docs

This section turns the official Spark and Databricks documentation into separate questions, grouped by source, so that you can revise one point at a time. Each heading links to the original page. Settings and version numbers change between releases, so check the docs for your own Spark or Databricks Runtime version.

Spark SQL performance tuning

Q1. How do you cache a table in Spark SQL, and how do you check it?
Use spark.catalog.cacheTable("t") or df.cache(), and release it with uncacheTable or df.unpersist(). Check with spark.catalog.isCached("t") or the Storage tab. Spark stores cached tables in columnar form and picks compression per column (spark.sql.inMemoryColumnarStorage.compressed, on by default). A larger batchSize (default 10,000) can compress better but risks OOM.

Q2. What does spark.sql.files.openCostInBytes do?
It is the estimated cost of opening a file (default 4 MB), used when Spark packs small files into one read partition. Each file counts as at least this many bytes, so many tiny files do not all land in one partition. The docs say overestimating is better.

Q3. How do you limit the number of partitions when reading files?
spark.sql.files.maxPartitionBytes (128 MB) sets the size of each partition. spark.sql.files.minPartitionNum suggests a minimum, and spark.sql.files.maxPartitionNum (Spark 3.5+) suggests a maximum, which makes Spark rescale when there are too many.

Q4. How do you control partitions from SQL?
Use partitioning hints, which match the DataFrame methods:

SELECT /*+ COALESCE(3) */ * FROM t;
SELECT /*+ REPARTITION(3, c) */ * FROM t;
SELECT /*+ REPARTITION_BY_RANGE(3, c) */ * FROM t;
SELECT /*+ REBALANCE(c) */ * FROM t;        -- needs AQE
-- newer Spark only. Spark 3.5 ignored this with no error:
SELECT /*+ REBALANCE_BY_SIZE('128m', c) */ * FROM t;

Q5. Where do Spark's row and size estimates come from?
From three places: the data source (such as Parquet metadata), the catalog (updated by ANALYZE TABLE), and the runtime (computed by AQE). Missing or wrong statistics lead to poor plans. In our test, open-source Delta Lake 3.2 rejected ANALYZE TABLE, so that route depends on your table format. Inspect them with DESCRIBE EXTENDED, EXPLAIN COST or df.explain(mode="cost").

Q6. When does Spark broadcast a table automatically?
When its estimated size is under spark.sql.autoBroadcastJoinThreshold (10 MB by default; -1 disables it). A broadcast that takes too long fails after spark.sql.broadcastTimeout (300 seconds).

Q7. What are the join hints, and which one wins?
BROADCAST, MERGE, SHUFFLE_HASH and SHUFFLE_REPLICATE_NL. If hints on the two sides conflict, that is the priority order. In our Spark 3.5 test, BROADCAST beat MERGE and MERGE beat SHUFFLE_HASH; with two hints on the same table, the first one won. A hint is not a guarantee, because some strategies do not support every join type.

SELECT /*+ BROADCAST(r) */ * FROM src s JOIN records r ON s.key = r.key;

spark.table("src").join(spark.table("records").hint("broadcast"), "key")

Q8. How does AQE change a join while it runs?
If one side turns out small after its shuffle, AQE replaces the sort merge join with a broadcast hash join. Both sides were already shuffled, but the sort and merge are saved, and spark.sql.adaptive.localShuffleReader.enabled reduces network traffic. In our test on Spark 3.5, the join was planned as SortMergeJoin and finished as BroadcastHashJoin.

Q9. When does AQE treat a partition as skewed?
Only if it is larger than both 5 times the median partition size (skewedPartitionFactor) and 256 MB (skewedPartitionThresholdInBytes). Spark then splits the skewed tasks and replicates the other side where needed.

Q10. What is a Storage Partition Join?
A way to avoid the shuffle in a join when both tables are already partitioned compatibly in a V2 source such as Iceberg. If it applies, the plan has no Exchange nodes before the join.

Q11. What does coalescePartitions.parallelismFirst do?
When true (the default), AQE ignores the advisory size while merging and only respects the minimum size, to keep parallelism high. The docs suggest setting it to false on a busy cluster, so that fewer small tasks are created.

Spark tuning guide

Q12. Java serialization or Kryo?
Kryo is often up to 10 times faster and more compact, but you should register your classes, or every object stores its full class name. Increase spark.kryoserializer.buffer for large objects. It matters most for RDDs and serialized caching in JVM code; DataFrames already use Tungsten's binary format.

Q13. How do you reduce memory used by data structures?
Prefer arrays and primitive types to collections like HashMap, avoid nested structures with many small objects, and use numeric IDs rather than strings as keys.

Q14. How do you find out how much memory a dataset uses?
Cache it and read the Storage page in the Spark UI. Java objects can take 2 to 5 times the space of the raw data, because of headers, pointers and boxed values.

Q15. What level of parallelism does Spark recommend?
2 to 3 tasks per CPU core. spark.default.parallelism applies to RDDs, while spark.sql.shuffle.partitions applies to DataFrames and SQL.

Q16. Why can a task hit OOM when the data fits in memory?
Because one task's working set, such as a large group in a groupBy, is too big. The simplest fix is more parallelism, so that each task has less input. Spark handles tasks as short as 200 ms well.

Q17. When should you use a broadcast variable?
For large read-only data such as a lookup table, instead of capturing it in the task closure. Spark logs each task's serialized size; tasks over about 20 KiB are worth optimizing.

Q18. What is data locality?
How close a task runs to its data. The levels, from best to worst, are PROCESS_LOCAL, NODE_LOCAL, NO_PREF, RACK_LOCAL and ANY. Spark briefly waits for a free slot close to the data, and spark.locality.wait sets how long. Raise it only if tasks are long and locality is poor.

Q19. Why is listing input files slow, and what helps?
Object stores such as S3 are slow to list many directories. For Spark SQL sources, tune spark.sql.sources.parallelPartitionDiscovery.threshold and .parallelism.

Databricks: When to partition tables

Q20. Should a Delta table on Databricks be partitioned?
Under 1 TB, no. From 1 TB to 100 TB, use liquid clustering instead. At 100 TB or more, partitioning may help, but test liquid clustering first. Any partition should hold at least 1 GB.

Q21. Which columns make good partition columns?
Low-cardinality fields such as dates or locations. Timestamps and other high-cardinality fields are poor choices. Partition columns must be top-level, so complex types and struct fields are not allowed.

Q22. What is ingestion time clustering?
Unpartitioned Delta tables automatically get partition-like performance for time-based queries, with no setup. If you run many UPDATE or MERGE operations, use liquid clustering on a column that follows ingestion order, such as an event date, to keep that benefit.

Q23. Partitioning or Z-order?
Z-order suits high-cardinality fields and needs OPTIMIZE. It only combines files within a partition, and you cannot Z-order a partition column. For new tables, Databricks recommends liquid clustering over both.

Q24. Why is over-partitioning risky?
It can slow queries, and fixing it may mean rewriting all the data. Hive-style folders are also not part of the Delta protocol, so workloads should not depend on them.

Databricks: Liquid clustering

Q25. When is liquid clustering useful?
Databricks recommends it for all new tables. It helps most with high-cardinality filters, heavy skew, fast-growing tables, concurrent writes and changing query patterns.

Q26. How do you enable, change and remove clustering?
Use the statements below. It cannot be combined with partitioning or ZORDER. To convert a partitioned Delta table, use the last statement (Databricks Runtime 18.1+).

CREATE TABLE sales (id BIGINT, order_date DATE) CLUSTER BY (order_date);
ALTER TABLE sales CLUSTER BY (order_date, region);
ALTER TABLE sales CLUSTER BY NONE;

ALTER TABLE old_sales REPLACE PARTITIONED BY WITH CLUSTER BY (order_date);

Q27. How do you choose clustering keys?
Use up to four columns that queries filter on most. Key order does not matter. Keys need statistics, and by default only the first 32 columns get them. For tables under 10 TB, extra keys can slow single-column filters. When migrating, start from your partition or ZORDER columns.

Q28. What does CLUSTER BY AUTO need?
Unity Catalog managed tables, Databricks Runtime 15.4 LTS+ for Delta, and predictive optimization. It may choose no keys for small or rarely queried tables. Watch out: CREATE OR REPLACE TABLE without CLUSTER BY AUTO turns it off.

Q29. Does every write cluster the data?
No. Clustering on write applies to operations such as INSERT, CTAS, COPY INTO from Parquet, and appends, and only above a size threshold. So you still need OPTIMIZE.

Q30. OPTIMIZE or OPTIMIZE FULL?
OPTIMIZE is incremental, rewriting only what needs clustering. Without predictive optimization, schedule it; every 1 to 2 hours suits heavily updated tables. OPTIMIZE FULL (Runtime 16.0+) reclusters everything and can take hours, so run it after first enabling clustering or changing keys.

Q31. Does changing keys rewrite existing data?
No. Only later writes and OPTIMIZE runs use the new keys. Use OPTIMIZE FULL to rewrite old data.

Q32. What is the compatibility catch?
Clustered Delta tables use reader version 3 and writer version 7. Older clients cannot read them, and the protocol cannot be downgraded.

Databricks: Adaptive query execution

Q33. What are AQE's four main optimizations?
It switches sort merge joins to broadcast joins, coalesces small shuffle partitions, handles skew in sort merge and shuffle hash joins, and detects empty relations. It uses real sizes measured at the end of each shuffle or broadcast stage.

Q34. When does AQE apply?
To non-streaming queries with at least one exchange (join, aggregate or window) or a subquery. The final plan can still match the original one.

Q35. What can AQE not do?
It never fixes skew in broadcast joins and does not reorder joins at runtime. For join order, rely on table statistics and the cost-based optimizer.

Q36. How can you see that AQE worked?
Look for an AdaptiveSparkPlan node, whose isFinalPlan flag becomes true when the query ends. Then look for a changed join node, a shuffle reader marked coalesced, a SortMergeJoin with isSkew true, or an empty LocalTableScan. EXPLAIN SQL does not run the query, so it shows only the first plan. In our Spark 3.5 test it still showed SortMergeJoin after the run had switched.

Q37. What is the broadcast threshold under AQE on Databricks?
spark.databricks.adaptive.autoBroadcastJoinThreshold defaults to 30 MB, higher than the 10 MB static threshold of open-source Spark.

Q38. Why was a small table not broadcast?
The join type may not allow broadcasting that side, such as the left side of a LEFT OUTER JOIN. Or too few of its partitions were non-empty, compared with spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin.

Q39. Should you still use broadcast hints with AQE?
Yes, when you know your data. A broadcast planned at the start is usually better than one chosen by AQE, which can happen only after both sides have shuffled. AQE respects hints.

Q40. Which join types get skew handling?
Left side: INNER, CROSS, LEFT OUTER, LEFT SEMI, LEFT ANTI. Right side: INNER, CROSS, RIGHT OUTER. FULL OUTER: neither side.

Q41. Should you set minPartitionNum?
No. It defaults to twice the cluster cores, and setting it overrides minPartitionSize. Databricks advises against it.

Q42. Does AQE work with streaming?
No. Also, in Structured Streaming, spark.sql.shuffle.partitions cannot change between restarts from the same checkpoint.

Databricks: Skew and spill

Q43. How do you detect skew on the stage page?
Open Summary Metrics and compare Max task duration with the 75th percentile. They should be about the same. If Max is 50% or more above the 75th percentile, the stage might have skew.

Q44. How do you detect spill?
The stage details at the top of the stage page show spill statistics when spill occurs. If none appear, there is no spill.

Q45. In what order should you fix spill?
Set spark.sql.shuffle.partitions to auto; reduce rows, columns or large values; use larger instances, with fewer, larger workers for complex ETL; and for streaming, limit the input rate.

Q46. In what order should you fix skew?
First check if AQE already handles it. Then lower the two skew thresholds if partitions fall below them, salt the keys, use liquid clustering if the skew comes from table layout, and use REBALANCE when writing.

Q47. What if a slow stage has neither skew nor spill?
Go back to the job through Associated Job Ids, and check whether the stage is limited by I/O.

Databricks: Spark configuration

Q48. Should you copy Spark settings to Databricks?
Generally no. Databricks recommends against setting most properties, especially when migrating from open-source Spark or upgrading, because old values override newer defaults.

Q49. Where can you set Spark properties?
On the compute (all notebooks and jobs on it), in a notebook (that SparkSession only), in Databricks SQL through admin settings or session SET, and at pipeline, compute or flow level in Lakeflow.

spark.conf.set("spark.sql.ansi.enabled", "true")  # this session only
spark.conf.get("spark.sql.shuffle.partitions")       # read a value

Q50. What can you set on serverless?
Only a small set of properties:

Q51. What can you not set?
The spark.dynamicAllocation.* properties: use Databricks autoscaling instead. spark.master and the driver host and port are managed by the platform on serverless and Lakeflow. spark.jars is not supported there either.

Databricks Runtime 19 release notes

Q52. What does Runtime 19 run on?
Apache Spark 4.2.0, Delta Lake 4.2.0, JDK 21 only, Python 3.12.3 and Scala 2.13. It reached general availability on July 23, 2026.

Q53. What changed for shuffle partitions?
Auto-optimized shuffle v2 is on by default when you do not set a fixed number. Workloads with a fixed count are unchanged, so old hardcoded values stop you from getting it.

Q54. Which performance changes matter most?
Automatic change data feed is GA and computes changes at query time, which Databricks reports makes MERGE and UPDATE about 15% faster on tables queried for changes. Array functions such as filter and transform can use code generation, and regular Python UDFs now use Arrow by default, which can change type coercion.

Q55. What should you check before upgrading?
JDK 17 is gone, so remove any JNAME setting. About 90 standard packages were removed, including plotly, seaborn, openai and pyodbc, so pin your own. Query behaviour changed too: HAVING runs before window functions in queries with generators, and JDBC pushdown now sends a real NULL rather than the string 'null'.

Databricks: What is Photon?

Q56. What is Photon?
A native C++ vectorized engine that processes columnar batches instead of using the JVM. Catalyst still plans the query, no code changes are needed, and anything unsupported falls back to the Spark runtime.

Q57. Which workloads gain the most from Photon?
Long queries over large data: joins, shuffles, scans and writes, including wide tables. Queries under two seconds gain little.

Q58. What does Photon not support?
UDFs, RDD and Dataset APIs, and stateful streaming. Those parts run on the Spark runtime or are not supported.

Q59. How do you enable Photon, and what does it cost?
It is on by default on serverless, SQL warehouses and, per the docs, classic compute. Use the Use Photon Acceleration checkbox or runtime_engine set to PHOTON. Photon instances use DBUs at a different rate.

Q60. How can you see which parts ran on Photon?
On classic compute, the SQL/DataFrame tab of the Spark UI shows Photon operators in orange. On SQL warehouses and serverless, the query profile shows the percentage of task time in Photon.

Q61. Which features need Photon?
Predictive I/O, and dynamic file pruning in MERGE, UPDATE and DELETE.

Databricks KB: Photon ran out of memory

Q62. Why does a query fail with "Photon ran out of memory"?
Photon uses table statistics to size its memory, so stale statistics can under-allocate. The risk rises with many joins, subqueries and aggregations, or very large data. Refresh statistics, split the query into smaller steps, and move to Runtime 13.3 LTS or later.

ANALYZE TABLE sales COMPUTE STATISTICS;

Databricks performance best practices

Q63. Vertical or horizontal scaling?
Vertical means a bigger machine, usually with a restart, and it has limits. Horizontal means more nodes, which only scales linearly when tasks are independent, since exchanging data between nodes adds network cost.

Q64. Does a larger cluster cost more?
For workloads that scale linearly, about the same, because you pay for duration and it finishes sooner. If cost matters more than speed, autoscaling is usually cheapest but not always fastest. Small datasets can even run slower on a distributed system.

Q65. Disk cache or Spark cache?
Prefer the disk cache, which keeps copies of remote files on local disks and updates automatically. Spark cache can hold any subquery and non-Parquet formats, but misuse can exhaust memory, so the rule of thumb is to avoid it.

Q66. What should you do about Python UDFs?
Use a native Spark function if one exists, because moving data between Python and Spark is slow. If none exists, use a pandas UDF, since Arrow makes the transfer efficient.

Q67. What are the ways to handle small files?
OPTIMIZE compacts them on demand, auto compact merges them after a write, and optimized writes produce better sizes while writing.

Q68. What does predictive optimization do?
For Unity Catalog managed tables, it uses query patterns to run maintenance automatically, including OPTIMIZE and ANALYZE. Databricks recommends enabling it, and turning off your own scheduled OPTIMIZE jobs.

Q69. Large files or small files?
Large files suit scan-heavy queries. Small files suit point lookups, since less data is read. Keep data in a natural time order during ingestion when you can.

Q70. How should you test performance?
Use production data or data with the same volume, file layout and skew. The first query on a cluster is slower, so test cold and warm runs separately.

Q71. What should you monitor?
Enable system tables and cluster log delivery for Spark event logs. Use Query Profile for per-task time, rows and memory, and set job alerts for SLA breaches.

13. Delta Lake

Delta Lake is an open-source storage layer that sits on top of cloud object storage (e.g. S3, ADLS) and adds reliability and performance to a plain Parquet data lake — readable by any compatible engine and built to work natively with Spark.

• ACID transactions — concurrent readers and writers never see a partial or corrupt table.
• Schema enforcement & evolution — writes that don't match the table schema fail fast, unless you explicitly opt in to merge a schema change.
• Time travel — query a table as of a prior version or timestamp, via its transaction log.
• Unified batch + streaming — a Delta table can be a sink and a source for both batch jobs and Structured Streaming at once.

df.write.format("delta").mode("overwrite").save("/data/delta/orders")

spark.read.format("delta").option("versionAsOf", 12).load("/data/delta/orders")

from delta.tables import DeltaTable
dt = DeltaTable.forPath(spark, "/data/delta/orders")
dt.update(condition="status = 'pending'", set={"status": "'shipped'"})
dt.vacuum(168)  # drop unreferenced files older than 168 hours

14. Incremental Ingestion

Incremental ingestion loads only the files that have arrived since the last run, instead of reprocessing everything. Databricks offers two mechanisms:

-- COPY INTO
COPY INTO my_table
FROM '/path/to/files'
FILEFORMAT = CSV
FORMAT_OPTIONS ('delimiter' = '|', 'header' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true');

# Auto Loader
spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "json") \
    .option("cloudFiles.schemaLocation", "/schema/orders") \
    .load("/path/to/files") \
  .writeStream \
    .option("checkpointLocation", "/checkpoints/orders") \
    .option("mergeSchema", "true") \
    .table("orders")

15. Structured Streaming

Structured Streaming treats a stream as an unbounded table: the same DataFrame API used for batch works here, with the engine re-running the query incrementally as new data arrives.

stream_df = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("subscribe", "orders") \
    .load()

query = stream_df.selectExpr("CAST(value AS STRING)") \
    .writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/checkpoints/orders") \
    .trigger(processingTime="1 minute") \
    .start("/data/delta/orders")

query.awaitTermination()

16. Memory & Caching

Each executor's JVM heap splits into a storage region (cached DataFrames/RDDs) and an execution region (shuffles, joins, sorts) that can borrow from each other under Spark's unified memory manager.

from pyspark import StorageLevel
df.persist(StorageLevel.MEMORY_AND_DISK)
df.unpersist()  # release cached partitions when done

17. Garbage Collection

Garbage collection (GC) is how the JVM automatically frees memory held by objects that are no longer used. Spark's driver and executors are JVM processes, so GC runs inside every job. You never call it yourself, but it can decide whether a job takes minutes or hours, because GC can pause the threads running your tasks.

This section builds up in order: where the memory goes, how GC cleans it, how a pause harms a job, how to detect that, and how to fix it, using real measurements.

Why Spark creates so much garbage

Say a task reads a partition and sums amounts per region. Every row it touches can create short-lived objects: strings, numbers, hash-map entries and buffers. The task finishes with them almost immediately, but the memory is not returned until GC runs. With 50 tasks on 10 executors processing millions of rows, this happens constantly. That is normal, and the JVM is built for it. The trouble starts only when objects stay alive.

Where the memory goes: the JVM heap

The JVM bets that most objects die young, so it splits the heap into two areas and cleans them differently.

 Executor JVM heap
 ┌───────────────────────────────┬───────────────────────────────┐
 │ YOUNG generation              │ OLD generation                │
 │  Eden │ Survivor │ Survivor   │  long-lived objects:          │
 │  new objects start here       │  cached data, large buffers   │
 └───────────────────────────────┴───────────────────────────────┘
   Minor GC: frequent, fast          Major / Full GC: slower
   Most objects die here             Survivors end up here

How Spark divides that heap

Spark then divides the same heap for its own use. Here is what --executor-memory 16g gives you with default settings:

 Heap: 16 GB
 ├─ Reserved                        300 MB
 ├─ Unified (16 GB - 300 MB) x 0.6  9.4 GB   spark.memory.fraction
 │    ├─ Storage: cached data       they share the 9.4 GB
 │    └─ Execution: shuffles, joins, sorts
 └─ User memory: remaining 40%      6.3 GB   your objects, metadata

 Outside the heap, so GC does not scan them:
 ├─ Memory overhead, max(10%, 384 MB)  1.6 GB   JVM internals, Python
 └─ Off-heap, 0 by default                      spark.memory.offHeap.size

This explains how Spark features connect to GC. Cached data stays alive for a long time, so it moves to the old generation and adds work for major GCs. Execution memory holds the shuffle and sort buffers. Spark stores most of those rows in Tungsten's compact binary format, so a billion rows do not become a billion objects. Operations that hold real JVM objects, such as collect_list, lose that benefit.

What a GC problem looks like, step by step

1. Tasks allocate objects in Eden. When Eden fills, a minor GC removes the dead objects and moves survivors along. This is quick.

2. Some objects keep surviving. A large aggregation or a cached table holds objects for a long time, so they get promoted to the old generation.

3. The old generation fills up. A major GC runs. With a stop-the-world collector, every task on that executor pauses, not just the one that caused it. That is why one executor with 5 running tasks counts as 5 seconds of GC Time for a single 1-second pause.

4. GC frees very little. The objects are still in use, so the heap is full again shortly after. The JVM repeats the collection, spending more time cleaning than working.

5. The job starts to fail. During long pauses the executor cannot send heartbeats, so the driver can mark it as lost and re-run its tasks. Eventually you see GC overhead limit exceeded or Java heap space.

Which collector is running?

How to detect it

GC logs show what the metrics cannot. They are written to each executor's stdout, not the driver's:

# Java 11 and later
--conf spark.executor.extraJavaOptions="-Xlog:gc*"

# Java 8
--conf spark.executor.extraJavaOptions="-verbose:gc -XX:+PrintGCDetails"

Many full GCs before a task finishes means tasks do not have enough memory. Many minor GCs but few major ones means the young generation is too small for the rate of allocation.

Measured: what actually changes GC

We ran the same aggregation with one setting changed at a time, and measured the GC share from Spark's own metrics. Each row is the average of two runs of 12 million rows and 20,000 groups, on Spark 3.5 and Java 8 with 4 local cores.

A second run used 40 million rows to find out when memory matters:

What these numbers teach:

• The query matters most. Swapping collect_list for sum nearly halved the time. No setting came close. It is the same reason built-in functions beat UDFs.
• More memory is not a speed-up by itself. Too little memory crashed the job, but once it fit, a larger heap was slower here, most likely because the collector had far more live objects to work through. Memory fixes failures; it does not automatically fix slowness.
• Do not chase the GC percentage. G1 cut GC from 13% to 3%, yet the job took about the same time, probably because G1's concurrent work uses CPU that this metric does not count. Raising shuffle partitions also lowered the share while doubling the runtime. Judge a change by total job time.
• Extra columns cost time. One more column held in the list added over 40%, so selecting only what you need is real GC tuning.

These are small heaps on one machine, so treat them as direction, not as universal rules. A cluster with large heaps and many executors can respond differently, particularly to G1, so measure your own job.

How GC relates to memory spills and OOM

Spills, GC and out-of-memory errors (OOM) look like three separate problems, but they are usually three stages of the same one: a task needs more memory than it can get. What differs is who responds, and each response costs more than the last.

The chain of events

Follow one task through a large sort or aggregation. Each step happens because the one before it did not solve the problem.

1. The task asks Spark for memory. Before filling a sort or hash buffer, the task requests execution memory from Spark's memory manager. If there is room, it continues in memory, which is the fast path.

2. Spark says no, so the task spills. If execution memory is exhausted, Spark writes the buffer to disk and frees it. This is a deliberate safety valve: the job gets slower, but it keeps running. Each active task can get at most about 1/N of execution memory, where N is the number of running tasks, so a big task can spill even when the executor seems to have room.

3. Meanwhile, the heap keeps filling. Serializing to disk creates objects of its own, and not everything lives in Spark's managed buffers. UDF results, deserialized rows, collect_list arrays and user data structures are ordinary JVM objects. The JVM responds with GC.

4. GC works harder and frees less. Live objects cannot be freed, so collections run more often and recover less. GC Time climbs. This is your warning that the job is close to failing.

5. OOM. A new object is needed and GC cannot make room for it, so the JVM throws OutOfMemoryError. In a similar way, a long pause can stop heartbeats, and the executor is lost before any error appears.

Task needs memory
   │
   ▼
Spark execution memory available?
   ├─ yes ─► keep working in memory            fast
   └─ no  ─► SPILL to disk and continue        slower, safe
                │
                ▼
Heap filling with objects Spark cannot spill?
   ├─ no  ─► job finishes
   └─ yes ─► GC runs more often                slower, warning
                │
                ▼
GC still cannot free enough?
   ├─ no  ─► job survives, but is slow
   └─ yes ─► OutOfMemoryError                  failure

Why spilling does not always prevent OOM

A common belief is that Spark spills to disk, so it cannot run out of memory. Spilling only protects data Spark can move. Some data has to be in memory all at once: a single huge record, a broadcast table, results returned by collect(), or one key's complete list. No amount of disk helps with those.

Measured: spills, GC and OOM together

These runs used 12 million rows on Spark 3.5 with 4 local cores. The skewed data put 80% of the rows on one key, so the row count stayed the same. The skewed 1 GB case ran 6 times; the others ran once.

Spill (memory) is the data's size in memory before spilling, and Spill (disk) is its compressed size on disk, which is why they differ. The crashes reported java.lang.OutOfMemoryError: Java heap space.

• Removing a spill is not the same as making a job faster. Tripling the memory ended the spill in the window sort, but the time barely changed and GC rose from 35% to 59%. The cost moved from disk to GC.
• Spilling did not save the skewed job. It spilled as much as the job that passed, and still crashed half the time, most likely because one key's list must fit in memory as a whole.
• High GC is the warning before OOM. The skewed runs that survived were spending over 60% of task time on GC, right on the edge of failing. A job that passes today can fail tomorrow with slightly more data.
• Skew can be the real cause. Same rows, same heap, same query: even data was stable at 13% GC. Only the distribution changed. Memory stopped the crash, but it was the slowest run, so fixing the skew is the better answer.

Reading the different OOM errors

"OOM" covers several failures, and they have different fixes. Before changing anything, find out where the memory ran out.

The container-killed case has nothing to do with GC, since the memory is outside the heap. Exit code 137 on Kubernetes usually means the same thing. Check for this before raising --executor-memory, which would not help.

From symptom to fix

# More memory for each shuffle task: split the same data more finely
spark.conf.set("spark.sql.shuffle.partitions", 400)

# Let AQE split skewed partitions in joins
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

Earlier we measured 200 partitions taking twice as long as 8 on a small dataset, so raise this only when partitions really are too large. Aim for shuffle partitions that fit comfortably in each task's share of memory.

On Databricks

The relationship is the same, but check two more things. In the Spark UI or query profile, look for spill on the slow stage. Photon keeps its memory outside the JVM heap, so a Photon job shows little JVM GC and fails differently, with a Photon ran out of memory error. Databricks suggests refreshing statistics with ANALYZE TABLE ... COMPUTE STATISTICS and breaking up complex queries. For heavy spill or OOM, more memory per instance also helps, as it does with any executor.

How to tune it, in order

The numbers above give the order: reduce the memory the job needs first, then resize, and change collector flags last.

from pyspark.sql import functions as F

# Slower: holds every value as a JVM object just to sum it afterwards
df.groupBy("region").agg(F.aggregate(F.collect_list("amount"), F.lit(0.0),
                                     lambda acc, x: acc + x).alias("total"))

# Faster: Spark keeps a running total in binary form
df.groupBy("region").agg(F.sum("amount").alias("total"))

# Keep the cached table small, and release it when finished
slim = df.select("region", "amount").filter(F.col("amount") > 0).cache()
slim.count()
# ... use it ...
slim.unpersist()

GC settings for spark-submit

GC_OPTS="-XX:+UseG1GC -XX:InitiatingHeapOccupancyPercent=35"
GC_OPTS="$GC_OPTS -XX:G1HeapRegionSize=16m"

spark-submit \
  --executor-memory 16g \
  --executor-cores 5 \
  --conf spark.executor.memoryOverhead=2g \
  --conf spark.memory.offHeap.enabled=true \
  --conf spark.memory.offHeap.size=4g \
  --conf spark.executor.extraJavaOptions="$GC_OPTS" \
  --conf spark.driver.extraJavaOptions="-XX:+UseG1GC" \
  sales_job.py

Off-heap memory is not free: add it to your container size. And never put -Xmx in extraJavaOptions, because Spark rejects it. Set the heap with --executor-memory.

Garbage collection on Databricks

The same JVM rules apply on Databricks, but the platform changes what you should do about them.

On classic compute, you can see GC in the cluster's Spark UI under the Executors tab, and read executor stdout through Spark UI or cluster log delivery. Flags go under Compute > Advanced options > Spark config:

spark.executor.extraJavaOptions -XX:+UseG1GC -Xlog:gc*
spark.driver.extraJavaOptions -XX:+UseG1GC

A typical troubleshooting story

A nightly Databricks job has gone from 40 minutes to 3 hours, with occasional lost executors. Here is how the steps above fit together.

1. Check the Spark UI. The Executors tab shows GC Time at about a third of Task Time, so GC is involved.

2. Check the slow stage. All tasks are high, not just one, which rules out skew and points to memory pressure across the executors.

3. Find the cause. The Storage tab shows a wide table cached early in the job and never released. It sits in the old generation for the whole run.

4. Fix the job, not the flags. Select only the needed columns before caching, and call unpersist() once the table has been used.

5. Compare total runtime. If the job is back to normal, stop. Only move on to Photon, a bigger instance or GC flags if it is still slow.

This story is an illustration of the method, not a measured result.

Common mistakes

Common interview questions

18. UDFs

Built-in SQL functions (§7) are always faster — Catalyst can optimize around them and Tungsten can codegen them. Reach for a UDF only when no built-in covers the logic.

from pyspark.sql.functions import udf, pandas_udf
from pyspark.sql.types import StringType

@udf(returnType=StringType())
def classify(amount):
    return "high" if amount > 1000 else "low"

df.withColumn("tier", classify(col("amount")))

# Pandas UDF: vectorized, batches rows through pandas/Arrow for far better throughput
@pandas_udf(StringType())
def classify_vectorized(amount: "pd.Series") -> "pd.Series":
    return amount.apply(lambda a: "high" if a > 1000 else "low")

19. Databricks Platform

• Notebooks — multi-language (Python/SQL/Scala/R) cells in one document, with results and visualizations inline; the primary authoring surface for both jobs and exploration.
• Clusters — all-purpose clusters for interactive notebook work; job clusters spun up per scheduled run and torn down after, to avoid paying for idle capacity.
• Unity Catalog — a single governance layer (permissions, lineage, audit) across every workspace's tables, views and ML models.
• Photon — a native-code query engine that accelerates SQL and DataFrame workloads without changing application code.
• Jobs / Workflows — schedules notebooks or JARs as production pipelines, with retries, alerting and task dependencies between steps.

20. Sizing a Cluster: 10 GB, 100 GB, 1 TB

Size a cluster by working through a chain: data size → tasks → cores → memory → instance type and count. You never need memory equal to the data size: Spark processes partitions in waves and spills to disk, so cluster size sets speed, not feasibility.

• Tasks ≈ input size / 128 MB. (1 GB = 1,024 MB in the binary units Spark uses, so 10 GB = 10,240 MB ≈ 80 tasks.)
• Cores: aim for 2–4 tasks per core; more cores than tasks just sit idle.
• Memory: about 4 GB per core for simple jobs, 6–8 GB per core for heavy joins.
• Instances (AWS): m family = 4 GB per vCPU, c = 2 GB, r = 8 GB. Sizes double: xlarge 4 vCPU, 2xlarge 8, 4xlarge 16. Prefer "d" variants (r5d) for local NVMe shuffle disks at scale.
• Executors: ~5 cores each; reserve 1 core and some memory per node for the OS and daemons; add ~10% memory overhead.

Worked example: why 2 × m5.2xlarge for 10 GB

• 80 tasks → target 12–16 worker cores; × 4 GB → ~48 GB executor memory.
• 2 × m5.2xlarge = 16 vCPU / 64 GB raw → ~12 cores / 48 GB usable after the OS reserve, and it survives one node failing.
• Each worker: 2 executors × 3 cores, 12 GB + 2 GB overhead. The driver only plans and coordinates, so m5.xlarge is enough.
• At this size a single 16-core machine (local[16]) or DuckDB/Polars is often simpler. File format matters more than cluster size: gzip CSV cannot be split and CSV is often 5–10× slower than Parquet.

Example submit for 1 TB

spark-submit --master yarn --deploy-mode cluster \
  --driver-memory 16g --num-executors 30 --executor-cores 5 --executor-memory 34g \
  --conf spark.executor.memoryOverhead=4g \
  --conf spark.sql.shuffle.partitions=8000 \
  --conf spark.sql.adaptive.enabled=true \
  --conf spark.sql.adaptive.skewJoin.enabled=true \
  --conf spark.sql.autoBroadcastJoinThreshold=100m \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.dynamicAllocation.enabled=true job.py

On Databricks you pick only the node type and worker count (e.g. r5d.4xlarge, autoscale, Photon); executors are sized for you. Use spot workers with an on-demand driver to cut cost.

Tuning from the Spark UI

21. Cores & Parallelism

A core is a task slot: one task (one partition) runs on one core at a time. Total parallelism = executors × cores per executor, and cores in an executor share its memory. Cloud vCPUs are usually hyperthreads.

cores needed ≈ (number of tasks × average seconds per task) / target seconds

• Measure time per task on a sample run (Spark UI → Stages → task duration).
• While every core is busy, more cores means faster for roughly the same cost, since you pay for core-seconds.
• Startup, shuffles and skew do not speed up with more cores, so returns diminish.
• Other caps on parallelism: Kafka partition count, unsplittable files, memory per core, shuffle partitions.

22. Flink and Databricks

Databricks is Spark-native and has no supported Flink runtime. Installing apache-flink on a cluster only starts a local mini-cluster on the driver: not distributed, not fault tolerant.

The real gap is latency: Flink processes event-at-a-time (milliseconds) while Spark micro-batches add seconds. When milliseconds matter, run Flink on a managed service (Amazon Managed Service for Apache Flink, Confluent Cloud, EMR, Kubernetes) and share data through Kafka and Delta/Iceberg tables:

Kafka ──► Flink (managed, ms latency) ──► alerts / serving DB
   │              └──► Delta / Iceberg on S3 ──┐
   └─────────────────────────────────────────► Databricks: analytics, ML, Unity Catalog

23. Moving pandas Jobs to PySpark

Three migration levels: run the script unchanged (driver only), switch to the pandas API on Spark (import pyspark.pandas as ps, few changes, distributed), or rewrite with the native DataFrame API for the best performance.

txn = spark.read.table("fraud.silver.transactions")
profiles = spark.read.table("fraud.silver.customer_profiles")

daily = (txn.filter(F.col("status") == "SUCCESS")
            .withColumn("day", F.to_date("event_time"))
            .groupBy("account_id", "day")
            .agg(F.count("txn_id").alias("txns"), F.sum("amount").alias("total")))

(daily.join(F.broadcast(profiles), "account_id", "left")
      .write.mode("overwrite").saveAsTable("fraud.gold.daily_account_summary"))

• Avoid toPandas() / collect() on big data, Python row loops, plain Python UDFs, and relying on row order.
• Validate the migration by reconciling old and new outputs: row counts, sums and row hashes.

24. Databricks Components

DATABRICKS ACCOUNT: identity (SSO, SCIM, groups, service principals) · Unity Catalog metastores
  WORKSPACES (dev / test / prod …)
    CONTROL PLANE (Databricks-managed): web UI · notebooks · job scheduler · cluster manager
    COMPUTE PLANE: classic clusters & SQL warehouses in YOUR cloud account
                   or SERVERLESS compute (Databricks-managed)
  STORAGE: your S3 / ADLS / GCS — Delta tables · Unity Catalog volumes

Databricks renames products often (Workflows → Lakeflow Jobs, Delta Live Tables → Lakeflow Declarative Pipelines), so check current documentation.

25. Databricks Notebooks

A notebook is a collaborative web document of code and Markdown cells attached to compute. spark, dbutils and display() are built in, and a notebook can run as a task in a job.

A notebook exported as source is a plain .py file with markers:

# Databricks notebook source      ← first line marks the file as a notebook
# MAGIC %md
# MAGIC # Title
# COMMAND ----------               ← cell boundary
from lakehouse.silver import build_silver

Production practice: keep logic in tested Python modules, keep notebooks thin, parameterise with widgets, run on job clusters or serverless, and deploy with Asset Bundles.

26. Developing in VS Code

Production teams write code in an IDE, test it locally, package it, and deploy wheels or Python files with Asset Bundles. Notebooks are mainly for exploration.

from databricks.connect import DatabricksSession

spark = DatabricksSession.builder.profile("dev").getOrCreate()
spark.read.table("samples.nyctaxi.trips").limit(5).show()

27. Auto Loader in Depth

Auto Loader is the cloudFiles Structured Streaming source. It loads only new files from S3, ADLS, GCS or Unity Catalog volumes into Delta, exactly once, with schema inference and evolution.

df = (spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaLocation", "/Volumes/fraud_dev/bronze/checkpoints/schema_txn")
        .load("/Volumes/fraud_dev/bronze/landing/transactions"))

(df.writeStream
   .option("checkpointLocation", "/Volumes/fraud_dev/bronze/checkpoints/bronze_txn")
   .trigger(availableNow=True)          # process everything new, then stop
   .toTable("fraud_dev.bronze.transactions"))

Auto Loader and HDFS

Auto Loader is Databricks-only and works on cloud object storage, not on-prem hdfs://. On Hadoop, use the open-source Structured Streaming file source with checkpoints (maxFilesPerTrigger, cleanSource), NiFi ListHDFS/FetchHDFS, or Kafka with an HDFS sink. When migrating, copy HDFS to S3/ADLS incrementally (DistCp -update, DataSync) and then point Auto Loader at it.

28. Lakeflow

CREATE OR REFRESH STREAMING TABLE silver_transactions (
  CONSTRAINT valid_amount EXPECT (amount > 0)          ON VIOLATION DROP ROW,
  CONSTRAINT valid_id     EXPECT (txn_id IS NOT NULL)  ON VIOLATION FAIL UPDATE
) AS SELECT * FROM STREAM(bronze_transactions);

CREATE OR REFRESH STREAMING TABLE dim_customer;
CREATE FLOW customer_scd2 AS AUTO CDC INTO dim_customer
FROM STREAM(bronze_customers)
KEYS (customer_id)
SEQUENCE BY extract_date
STORED AS SCD TYPE 2;

29. Liquid Clustering

Liquid clustering is a Delta data-layout feature that replaces partitioning and Z-ORDER. You declare up to 4 clustering keys; Delta incrementally co-locates rows with similar key values so file-level min/max statistics skip most files. Keys can be changed later without rewriting the table.

CREATE TABLE fraud_prod.silver.transactions (...) CLUSTER BY (txn_date, account_id);
ALTER TABLE fraud_prod.silver.transactions CLUSTER BY (txn_date, channel);  -- new keys
ALTER TABLE fraud_prod.silver.transactions CLUSTER BY AUTO;                 -- Databricks chooses
OPTIMIZE fraud_prod.silver.transactions;          -- clusters new data incrementally
OPTIMIZE fraud_prod.silver.transactions FULL;     -- re-cluster everything
DESCRIBE DETAIL fraud_prod.silver.transactions;   -- shows clusteringColumns

Built in, but opt-in per table

• Nothing to install, but a table is clustered only if you declare CLUSTER BY (or CLUSTER BY AUTO).
• Clustering happens during OPTIMIZE: schedule it, or enable predictive optimisation on UC managed tables.
• It adds a table feature (protocol upgrade), so older readers and writers may lose access.
• Pick 1–4 keys from frequent filter and join columns; high-cardinality columns are fine.

30. Asset Bundles & Job Scheduling

A Databricks Asset Bundle defines jobs, pipelines and their deployment targets as code in databricks.yml. The same bundle is promoted from dev to prod.

fraud-lakehouse/
├── databricks.yml          bundle, include, artifacts (wheel), variables, targets
├── resources/*.job.yml     one file per job / pipeline
├── pyproject.toml  src/  tests/
└── .github/workflows/deploy.yml

resources:
  jobs:
    fraud_daily:
      schedule:
        quartz_cron_expression: "0 0 2 * * ?"
        timezone_id: Asia/Kolkata
        pause_status: UNPAUSED
      max_concurrent_runs: 1
      queue: { enabled: true }
      timeout_seconds: 7200
      email_notifications: { on_failure: ["${var.alert_email}"] }
      health:
        rules: [{ metric: RUN_DURATION_SECONDS, op: GREATER_THAN, value: 3600 }]
      tasks:
        - task_key: silver
          depends_on: [{ task_key: bronze }]
          python_wheel_task:
            package_name: fraud_lakehouse
            entry_point: silver
            parameters: ["--catalog", "{{job.parameters.catalog}}"]
          max_retries: 2

How scheduling works

• databricks bundle deploy -t prod creates or updates the job and its schedule. The Lakeflow Jobs scheduler in the control plane then starts runs, creates compute, runs the task DAG with retries and alerts, and terminates compute.
• Quartz cron fields: seconds minutes hours day-of-month month day-of-week, with ? in one day field. 0 0 2 * * ? = daily 02:00; 0 30 6 ? * MON-FRI = weekdays 06:30; 0 0 23 L * ? = last day of the month. Always set timezone_id.
• Other triggers: file arrival, table update, continuous (streaming), manual / API.
• Enterprise schedulers (Control-M, Autosys, Airflow) can trigger jobs through the API, e.g. DatabricksRunNowOperator.
• Only CI/CD deploys to prod; backfills run through job parameters on idempotent jobs; no manual UI edits.

31. Production Databricks Setup

resource "databricks_catalog" "prod_fraud" {
  name           = "prod_fraud"
  storage_root   = "s3://bank-uc-prod-fraud/managed"
  owner          = "grp-fraud-data-owners"
  isolation_mode = "ISOLATED"
}

resource "databricks_workspace_binding" "prod_fraud" {
  securable_name = databricks_catalog.prod_fraud.name
  workspace_id   = var.prod_workspace_id
}

32. Environments: Dev, Test, Preprod, Prod

Environments are separated by workspace, catalog and identity, not by cluster. The standard model is one workspace per environment (in separate cloud accounts) sharing a regional metastore, with catalogs such as dev_fraud … prod_fraud bound to their workspace. Smaller teams may share a workspace and separate by catalog only.

CREATE OR REPLACE TABLE preprod_fraud.silver.transactions
  DEEP CLONE prod_fraud.silver.transactions;      -- full, independent copy
CREATE OR REPLACE TABLE test_fraud.silver.transactions
  SHALLOW CLONE prod_fraud.silver.transactions;   -- metadata only, points at prod files

• Cluster policies per environment (managed with Terraform): dev = small, spot, auto-terminate, required tags; preprod/prod = job clusters, fixed runtime, allowed node types, cost-centre tags.
• One bundle, four targets: only variables (catalog, policy, host, run_as) change. Build once, deploy many.
• Avoid: a shared workspace without catalog binding, a preprod unlike prod, unmasked PII in lower environments, interactive clusters in prod.